package student.convert

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import util.RandDomData

import org.apache.flink.configuration.Configuration

object TestState {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.addSource(new RandDomData)
      .assignAscendingTimestamps(_.writeTime)
      .map(_.code)
      .keyBy(_=>true)
      .flatMap(new RichFlatMapFunction[String,Long] {
        private var valueState: ValueState[Long]=_
        override def open(parameters: Configuration): Unit = { // 使用正确的 Configuration 类
          valueState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("ValueState", classOf[Long]))
        }
        override def flatMap(in: String, collector: Collector[Long]): Unit = {
          valueState.update(valueState.value()+1)
          collector.collect(valueState.value())
        }
      })
      .print()

    env.execute()
  }
}
